package com.minelx.hellopandas.core;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import java.util.List;

import static java.util.stream.Collectors.toList;

public class PandasMethodProvider {

	private final PandasChief pandasChief;

	private PandasMethodProvider(PandasChief pandasChief) {
		this.pandasChief = pandasChief;
	}

	public DataFrameDescription dropDuplicates(DataFrameDescription file, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("file", file.dumpToJSON());
		params.put("destination", destinationURI);
		return pandasChief.doRequest("dropDuplicates", params);
	}

	public DataFrameDescription dropNaN(DataFrameDescription file, List<String> columns, String how, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("file", file.dumpToJSON());
		params.put("columns", columns);
		params.put("how", how);
		params.put("destination", destinationURI);
		return pandasChief.doRequest("dropNaN", params);
	}

	public DataFrameDescription retains(DataFrameDescription file, List<String> retainedColumns, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("file", file.dumpToJSON());
		params.put("retained", retainedColumns);
		params.put("destination", destinationURI);
		return pandasChief.doRequest("retains", params);
	}

	public DataFrameDescription aggregate(DataFrameDescription file, List<String> groupBy, JSONArray specs, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("file", file.dumpToJSON());
		params.put("groupBy", groupBy);
		params.put("specs", specs);
		params.put("destination", destinationURI);
		return pandasChief.doRequest("aggregate", params);
	}

	public DataFrameDescription drop(DataFrameDescription file, List<String> droppedItems, int axis, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("file", file.dumpToJSON());
		params.put("droppedItems", droppedItems);
		params.put("axis", axis);
		params.put("destination", destinationURI);
		return pandasChief.doRequest("drop", params);
	}

	public DataFrameDescription merge(DataFrameDescription left,
	                                  DataFrameDescription right,
	                                  List<String> leftOn,
	                                  List<String> rightOn,
	                                  String how, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("left", left.dumpToJSON());
		params.put("right", right.dumpToJSON());
		params.put("leftOn", leftOn);
		params.put("rightOn", rightOn);
		params.put("how", how);
		params.put("destination", destinationURI);
		return pandasChief.doRequest("merge", params);
	}

	public DataFrameDescription mergeAll(DataFrameDescription hostFile,
	                                     List<JoinOption> joinOptions,
	                                     String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("hostFile", hostFile.dumpToJSON());
		params.put("joinOptions", new JSONArray(joinOptions.stream().map(JoinOption::toJSONObject).collect(toList())));
		params.put("destination", destinationURI);
		return pandasChief.doRequest("mergeAll", params);
	}

	public DataFrameDescription concat(List<DataFrameDescription> files, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("concatPaths", files.stream()
				.map(DataFrameDescription::dumpToJSON)
				.collect(toList()));
		params.put("destination", destinationURI);
		return pandasChief.doRequest("concat", params);
	}

	public DataFrameDescription filter(DataFrameDescription file, JSONObject condition, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("file", file.dumpToJSON());
		params.put("condition", condition);
		params.put("destination", destinationURI);
		return pandasChief.doRequest("filter", params);
	}

	public DataFrameDescription list(JSONArray select, JSONObject from, JSONArray conditions, String destinationURI) {
		JSONObject params = new JSONObject();
		params.put("select", select);
		params.put("from", from);
		params.put("where", conditions);
		params.put("destination", destinationURI);
		// do list
		return pandasChief.doRequest("list", params);
	}

	public static PandasMethodProvider create(PandasChief pandasChief) {
		return new PandasMethodProvider(pandasChief);
	}
}
